package cn.tpshion.task.server;

import cn.tpshion.task.config.support.GoTaskProperties;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

@Slf4j
public class NettyClient implements InitializingBean, DisposableBean {

    private volatile EventLoopGroup group;
    private volatile boolean stopFlag = false;
    private final AtomicInteger retryCount = new AtomicInteger(0);

    private final GoTaskProperties properties;

    private final TaskChannelInitializer taskChannelInitializer;

    public NettyClient(GoTaskProperties properties, TaskChannelInitializer taskChannelInitializer) {
        this.properties = properties;
        this.taskChannelInitializer = taskChannelInitializer;
    }

    public void start() {
        stopFlag = false;
        retryCount.set(0);
        doConnect();
    }

    private void doConnect() {
        // 停止条件判断
        if (shouldStop()) {
            log.info("Stop reconnecting");
            return;
        }

        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
            .channel(NioSocketChannel.class)
            .handler(taskChannelInitializer);

        ChannelFuture future = bootstrap.connect(properties.getHost(), properties.getPort());
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                onConnectSuccess(f.channel());
            } else {
                onConnectFailure();
            }
        });
    }

    // 连接成功处理
    private void onConnectSuccess(Channel channel) {
        retryCount.set(0);
        log.info("Connected to server");
        // 添加连接关闭监听
        channel.closeFuture().addListener(f -> {
            if (!stopFlag) {
                log.error("Connection closed, trigger reconnect");
                scheduleReconnect();
            }
        });
    }

    // 连接失败处理
    private void onConnectFailure() {
        log.error("Connect failed");
        scheduleReconnect();
    }

    // 调度重连任务
    private void scheduleReconnect() {
        if (shouldStop()) return;

        log.info("Scheduling reconnect in {}s (retry {}/{})",
            properties.getFixedDelay(), retryCount.incrementAndGet(), properties.getMaxRetry());

        group.schedule(() -> {
            if (!stopFlag) {
                // 关闭旧资源后重新连接
                group.shutdownGracefully();
                doConnect();
            }
        }, properties.getFixedDelay(), TimeUnit.SECONDS);
    }

    // 停止条件判断
    private boolean shouldStop() {
        return stopFlag || (properties.getMaxRetry() != -1 && retryCount.get() >= properties.getMaxRetry());
    }

    public void stop() {
        log.info("Stopping client");
        stopFlag = true;
        if (group != null) {
            log.info("Shutting down event loop group");
            group.shutdownGracefully();
        }
    }

    @Override
    public void destroy() {
        stop();
    }

    @Override
    public void afterPropertiesSet() {
        start();
    }

}
